# Copyright (c) 2023, NVIDIA CORPORATION. All rights reserved.

"""iFlytekSpark model."""

import torch
import os
from megatron import get_args
from megatron.core import mpu, tensor_parallel
from .module import MegatronModule, fp32_to_float16, float16_to_fp32

from .enums import AttnMaskType
from .language_model import parallel_lm_logits
from .language_model import get_language_model
from .utils import init_method_normal
from .utils import scaled_init_method_normal

from megatron.model import LayerNorm
from .language_model import EmbeddingPipe
from .transformer import ParallelTransformerLayerPipe
from deepspeed.pipe import PipelineModule, LayerSpec, TiedLayerSpec


def post_language_model_processing(lm_output, labels, logit_weights,
                                   parallel_output,
                                   fp16_lm_cross_entropy):

    # Output. Format [s b h]
    output = parallel_lm_logits(
        lm_output,
        logit_weights,
        parallel_output)

    if labels is None:
        # [s b h] => [b s h]
        return output.transpose(0,1).contiguous()
    else:
        # [b s] => [s b]
        labels = labels.transpose(0,1).contiguous()
        if fp16_lm_cross_entropy:
            assert output.dtype == torch.half
            loss = tensor_parallel.vocab_parallel_cross_entropy(output, labels)
        else:
            loss = tensor_parallel.vocab_parallel_cross_entropy(output.float(), labels)
        
        # [s b] => [b, s]
        loss = loss.transpose(0,1).contiguous()
        return loss


class iFlytekSparkModel(MegatronModule):
    """iFlytekSpark Language model."""

    def __init__(self,
                 config,
                 num_tokentypes=0,
                 parallel_output=True,
                 pre_process=True,
                 post_process=True,
                 return_moe_loss=True):
        args = get_args()
        super().__init__(config=config, share_embeddings_and_output_weights=not args.untie_embeddings_and_output_weights)

        self.parallel_output = parallel_output
        self.pre_process = pre_process
        self.post_process = post_process
        self.fp16_lm_cross_entropy = args.fp16_lm_cross_entropy
        self.return_moe_loss = return_moe_loss
        self.untie_embeddings_and_output_weights = args.untie_embeddings_and_output_weights

        self.language_model, self._language_model_key = get_language_model(
            config=config,
            num_tokentypes=num_tokentypes,
            add_pooler=False,
            encoder_attn_mask_type=AttnMaskType.causal,
            pre_process=self.pre_process,
            post_process=self.post_process,
            num_experts=args.num_experts)
        
        if not args.untie_embeddings_and_output_weights:
            self.initialize_word_embeddings()

    def set_input_tensor(self, input_tensor):
        """See megatron.model.transformer.set_input_tensor()"""
        self.language_model.set_input_tensor(input_tensor)

    def forward(self, input_ids, position_ids, attention_mask,
                retriever_input_ids=None,
                retriever_position_ids=None,
                retriever_attn_mask=None,
                labels=None, tokentype_ids=None, inference_params=None,
                curriculum_seqlen=None):
        args = get_args()
        if curriculum_seqlen is not None:
            args.curriculum_seqlen = curriculum_seqlen
            if curriculum_seqlen < input_ids.size()[1]:
                # seqlen-based curriculum learning
                # input_ids, position_ids, labels have size [batch size, seqlen]
                input_ids = input_ids[:, :curriculum_seqlen].contiguous()
                position_ids = position_ids[:, :curriculum_seqlen].contiguous()
                if labels is not None:
                    labels = labels[:, :curriculum_seqlen].contiguous()

                # attention_mask has size [1, 1, seqlen, seqlen]
                attention_mask = attention_mask[:, :, :curriculum_seqlen, :curriculum_seqlen].contiguous()
        else:
            if args.curriculum_learning_legacy:
                # If got a None input, need to reset curriculum_seqlen on user side
                args.curriculum_seqlen = args.seq_length

        lm_output, *moe_losses = self.language_model(
            input_ids,
            position_ids,
            attention_mask,
            retriever_input_ids=retriever_input_ids,
            retriever_position_ids=retriever_position_ids,
            retriever_attn_mask=retriever_attn_mask,
            inference_params=inference_params)

        if self.post_process:
            lm_output = post_language_model_processing(
                lm_output, labels,
                self.language_model.output_layer.weight if self.untie_embeddings_and_output_weights else self.shared_embedding_or_output_weight(),
                self.parallel_output,
                self.fp16_lm_cross_entropy)

        if self.return_moe_loss:
            return (lm_output, *moe_losses)
        else:
            return lm_output

    def state_dict_for_save_checkpoint(self, prefix='', keep_vars=False):

        state_dict_ = {}
        language_model_state_dict = self.language_model.state_dict_for_save_checkpoint(
                prefix=prefix, keep_vars=keep_vars)
        # MoE states need to be handled separately by DeepSpeed engine, thus
        # moving them to the top level dictionary
        if "moe_state_dict" in language_model_state_dict:
            for key in list(language_model_state_dict["moe_state_dict"].keys()):
                state_dict_[key] = language_model_state_dict["moe_state_dict"].pop(key)
            del language_model_state_dict["moe_state_dict"]
        state_dict_[self._language_model_key] = language_model_state_dict
        # Save word_embeddings.
        if self.post_process and not self.pre_process and not self.untie_embeddings_and_output_weights:
            state_dict_[self._word_embeddings_for_head_key] \
                = self.word_embeddings.state_dict(prefix=prefix,
                                                  keep_vars=keep_vars)
        return state_dict_

    def load_state_dict(self, state_dict, strict=True):
        """Customized load."""

        # Load word_embeddings.
        if self.post_process and not self.pre_process and not self.untie_embeddings_and_output_weights:
            self.word_embeddings.load_state_dict(
                state_dict[self._word_embeddings_for_head_key], strict=strict)
        # Gather MoE states and move under language model
        moe_state_dict = {}
        for key in list(state_dict.keys()):
            if 'expert' in key and 'moe.gate.wg.weight' not in key:
                moe_state_dict[key] = state_dict.pop(key)
        if self._language_model_key in state_dict:
            state_dict = state_dict[self._language_model_key]
        if len(moe_state_dict) > 0:
            state_dict["moe_state_dict"] = moe_state_dict
        args = get_args()
        if args.apply_lora:
            strict=False
        self.language_model.load_state_dict(state_dict, strict=strict)

def build_lora_model(model, config):
    args = get_args()
    model = apply_lora_layer(model, config)
    if args.lora_from_pretrained is not None:
        assert os.path.isdir(args.lora_from_pretrained)
        tp_rank = mpu.get_tensor_model_parallel_rank()
        pp_rank = mpu.get_pipeline_model_parallel_rank()
        ckpt_path = os.path.join(
            args.lora_from_pretrained,
            f'mp_rank_{tp_rank:02d}_model_states.pt'
        )
        from megatron import print_rank_0
        print_rank_0(f"loading lora pretrained weight from {args.lora_from_pretrained}")
        state_dict = torch.load(ckpt_path)
        if 'module' in state_dict:
            state_dict = state_dict['module']
        model.load_state_dict(state_dict, strict=False)
    
    from megatron.core.tensor_parallel.lora_layer import mark_only_lora_as_trainable
    mark_only_lora_as_trainable(model)
    
    return model

def apply_lora_layer(model, config):
    args = get_args()
    from megatron.core.tensor_parallel.lora_layer import (
        LoRAQKVParallelLinear, LoRAColumnParallelLinear, LoRARowParallelLinear
    )
    from megatron.core.tensor_parallel.layers import (
        ColumnParallelLinear, RowParallelLinear
    )
    
    for name, module in model.named_modules():
        if (args.adapt_q or args.adapt_k or args.adapt_v) and "self_attention.query_key_value" in name:
            assert isinstance(module, ColumnParallelLinear)
            lora_module = LoRAQKVParallelLinear(
                in_features=module.input_size,
                out_features=module.output_size,
                config=config,
                init_method=config.init_method,
                r=args.lora_rank,
                lora_alpha=args.lora_alpha,
                lora_dropout=args.lora_dropout,
                adapt_q=args.adapt_q,
                adapt_k=args.adapt_k,
                adapt_v=args.adapt_v,
                merge_weights=args.merge_weights,
                bias=module.bias is not None,
                partition_stride=module.stride,
                skip_bias_add=module.skip_bias_add,
                gather_output=module.gather_output,
                use_cpu_initialization=args.use_cpu_initialization,
            )
            lora_module.weight.data = module.weight.data
            lora_module.bias.data = module.bias.data
            update_module(model, name, lora_module)
        elif (args.adapt_k or args.adapt_v) and "self_attention.key_value" in name:
            raise RuntimeError(f"k_v_proj has not been supported yet")
        elif (args.adapt_o and "self_attention.dense" in name) or \
             (args.adapt_fc2 and "dense_4h_to_h" in name):
            assert isinstance(module, RowParallelLinear)
            lora_module = LoRARowParallelLinear(
                in_features=module.input_size,
                out_features=module.output_size,
                config=config,
                init_method=config.init_method,
                r=args.lora_rank,
                lora_alpha=args.lora_alpha,
                lora_dropout=args.lora_dropout,
                merge_weights=args.merge_weights,
                partition_stride=module.stride,
                skip_bias_add=module.skip_bias_add,
                input_is_parallel=module.input_is_parallel,
                use_cpu_initialization=args.use_cpu_initialization,
            )
            lora_module.weight.data = module.weight.data
            lora_module.bias.data = module.bias.data
            update_module(model, name, lora_module)
        elif (args.adapt_q and "self_attention.query" in name) or \
             (args.adapt_k and "self_attention.key" in name) or \
             (args.adapt_v and "self_attention.value" in name) or \
             (args.adapt_fc1 and "dense_h_to_4h" in name):
            assert isinstance(module, ColumnParallelLinear)
            lora_module = LoRAColumnParallelLinear(
                in_features=module.input_size,
                out_features=module.output_size,
                config=config,
                init_method=config.init_method,
                r=args.lora_rank,
                lora_alpha=args.lora_alpha,
                lora_dropout=args.lora_dropout,
                merge_weights=args.merge_weights,
                partition_stride=module.stride,
                skip_bias_add=module.skip_bias_add,
                gather_output=module.gather_output,
                use_cpu_initialization=args.use_cpu_initialization,
            )
            lora_module.weight.data = module.weight.data
            lora_module.bias.data = module.bias.data
            update_module(model, name, lora_module)
    
    return model


def _get_module_by_name(model, module_name):
    name_list = module_name.split(".")
    for name in name_list[:-1]:
        if hasattr(model, name):
            model = getattr(model, name)
        else:
            return None, None
    if hasattr(model, name_list[-1]):
        leaf_module = getattr(model, name_list[-1])
        return model, leaf_module
    else:
        return None, None


def update_module(model, module_name, new_module):
    super_module, leaf_module = _get_module_by_name(model, module_name)
    setattr(super_module, module_name.split('.')[-1], new_module)

def CrossEntropy(output, labels):
    labels, loss_mask = labels[0], labels[1]

    args = get_args()

    # [b s] => [s b]
    labels = labels.transpose(0, 1).contiguous()
    losses = tensor_parallel.vocab_parallel_cross_entropy(output.contiguous().float(), labels)
    # [s b] => [b, s]
    losses = losses.transpose(0, 1).contiguous()
    loss_mask = loss_mask.view(-1)
    loss = torch.sum(losses.view(-1) * loss_mask) / loss_mask.sum()
    return loss


class iFlytekSparkModelPipe(PipelineModule,MegatronModule):
    """iFlytekSpark Language model."""

    def __init__(self,
                 config,
                 num_tokentypes=0,
                 parallel_output=True):
        args = get_args()
        self.parallel_output = parallel_output

        if config.init_method is None:
            config.init_method = init_method_normal(config.init_method_std)

        if config.output_layer_init_method is None:
            config.output_layer_init_method = scaled_init_method_normal(config.init_method_std,
                                                                        config.num_layers)

        self.specs = []

        def _to_float16(inputs):
            if args.fp16:
                return fp32_to_float16(inputs, lambda v: v.half())
            elif args.bf16:
                return fp32_to_float16(inputs, lambda v: v.bfloat16())
            else:
                return inputs

        self.specs.append(_to_float16)

        # Embedding layer
        self.specs.append(TiedLayerSpec('embed',
                                        EmbeddingPipe,
                                        args.hidden_size,
                                        args.padded_vocab_size,
                                        args.max_position_embeddings,
                                        args.hidden_dropout,
                                        config,
                                        num_tokentypes=num_tokentypes,
                                        embedding_weights_in_fp32=args.embedding_weights_in_fp32,
                                        tied_weight_attr='word_embeddings_weight'))

        for layer_idx in range(args.num_layers):
            self.specs.append(
                LayerSpec(ParallelTransformerLayerPipe,
                    config,
                    layer_number=layer_idx,
                    self_attn_mask_type=AttnMaskType.causal))

        # Final layernorm after transformer layers
        self.specs.append(
            LayerSpec(LayerNorm,
                      args.hidden_size,
                      eps=args.layernorm_epsilon))

        def _logits_helper(embedding, lm_output):
            """A wrapper to massage inputs/outputs from pipeline. """
            return parallel_lm_logits(
                lm_output,
                embedding.word_embeddings_weight,
                self.parallel_output)

        self.specs.append(
            TiedLayerSpec('embed',
                          EmbeddingPipe,
                          args.hidden_size,
                          args.padded_vocab_size,
                          args.max_position_embeddings,
                          args.hidden_dropout,
                          config,
                          num_tokentypes=num_tokentypes,
                          embedding_weights_in_fp32=args.embedding_weights_in_fp32,
                          forward_fn=_logits_helper,
                          tied_weight_attr='word_embeddings_weight')
        )

        # Convert to fp32 if needed
        if args.fp16 or args.bf16:
            self.specs.append(float16_to_fp32)

        if args.checkpoint_activations:
            interval = args.checkpoint_num_layers
        else:
            interval = 0

        from deepspeed.runtime.pipe.topology import PipeModelDataParallelTopology
        topo = PipeModelDataParallelTopology(num_pp=mpu.get_pipeline_model_parallel_world_size(),
                                             num_mp=mpu.get_tensor_model_parallel_world_size(),
                                             num_dp=mpu.get_data_parallel_world_size())

        super().__init__(layers=self.specs,
                         loss_fn=CrossEntropy,
                         topology=topo,
                         activation_checkpoint_interval=interval,
                         partition_method='type:transformer')
